package com.atguigu.utils;

import com.atguigu.bean.MyAnnonation;
import com.atguigu.common.GmallConfig;
import lombok.SneakyThrows;
import org.apache.flink.connector.jdbc.JdbcConnectionOptions;
import org.apache.flink.connector.jdbc.JdbcExecutionOptions;
import org.apache.flink.connector.jdbc.JdbcSink;
import org.apache.flink.connector.jdbc.JdbcStatementBuilder;
import org.apache.flink.streaming.api.functions.sink.SinkFunction;

import java.lang.reflect.Field;
import java.lang.reflect.Method;
import java.sql.PreparedStatement;
import java.sql.SQLException;
//dws
public class MyClickHouseUtil {
    public static <T> SinkFunction<T> getSinkFunction(String sql){//T就是输入的类型，可以是任意对象，t就是输入的T实例对象，即每一条数据
      return  JdbcSink.sink(
              sql,
              new JdbcStatementBuilder<T>() {
                  @SneakyThrows
                  @Override
                  public void accept(PreparedStatement preparedStatement, T t) throws SQLException {
                      //todo 通过反射的方式获取属性值
                      Class<?> clz = t.getClass();//获取对象t的类型

                      //获取所有方法
//                      Method[] methods = clz.getMethods();//获取public修饰的方法
//                      for (Method method : methods) {
//                          method.invoke(t);
//                      }
                      //获取属性名
//                      Field[] fields = clz.getFields();//获取public修饰的属性名
                      Field[] fields = clz.getDeclaredFields();//获取所有属性名，包括private修饰的

                      int offset=0;
                      //遍历字段名，获取每个字段名对应的字段值并将其写入占位符中
                      for (int i = 0; i < fields.length; i++) {
                          Field field = fields[i];//获取下标为i的字段名
                          field.setAccessible(true);//强制获取权限，否则field.getAnnotation(MyAnnonation.class)和field.get(t)可能会报错

                          //todo 尝试获取字段上的注解，如果字段上有这个MyAnnonation注解，就跳过,比如javabean有10个属性，
                          // 有可能clickhouse只有8个字段，只要其中的8个属性对应的值写到clickhouse里，那么两个属性上就写
                          // 上@MyAnnonation，遇到这个注解，我们就要手动设置跳过，但是要注意占位符的下标问题
                          MyAnnonation myAnnotation = field.getAnnotation(MyAnnonation.class);//如果返回null，就说明属性上面没有这个注解
                          if(myAnnotation !=null){
                              //todo 说明该属性上面有这个注解
                            offset--;
                            continue;//todo 跳出当次循环

                          }
                          Object value = field.get(t);//获得属性名对应的属性值，由于设置了field.setAccessible(true)，这里的异常可以抛出去了
                          preparedStatement.setObject(i+1+offset,value);//占位符下标是从1开始

                      }



                  }
              },
              JdbcExecutionOptions.builder()
                      .withBatchSize(5)
                      .withBatchIntervalMs(1000L)
                      .build(),
              new JdbcConnectionOptions.JdbcConnectionOptionsBuilder()
                      .withDriverName(GmallConfig.CLICKHOUSE_DRIVER)
                      .withUrl(GmallConfig.CLICKHOUSE_URL)
                      .build()
      );
    }
}
